package cn.dmp.report

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}



/**
  * 统计日志文件中省市的数据量分布情况（将parquet文件数据统计后输出成json格式到磁盘,统计后存储到数据库中））
  *
  * 需求一：
  *     将统计的结果输出成json格式，并输出到磁盘目录。
  * 需求二：在ProCityRpt_ToMysql类里。
  *     将结果写到Mysql数据库中。
  * 需求三：
  *     利用spark算子的方式进行离线数据统计:还是统计各省市的数据量分布情况
  */
object ProCityRpt {
  def main(args: Array[String]): Unit = {

    if (args.length != 2){
      println(
        """
          |cn.dmp.report.ProCityRpt
          |参数：
          | logInputPath
          | resultOutputPath
        """.stripMargin)
      sys.exit()
    }

    // 1. 接受程序参数
    val Array(logInputPath, resultOutputPath) = args

    // 2. 创建sparkConf、sparkContext
    //val conf: SparkConf = new SparkConf().setAppName("ProCityRpt").setMaster("local[*]")

    /**
      * spark优化：spark.serializer修改序列化方式
      * 进行节点的数据传递，或者保存数据时都会进行序列化。
      * spark默认的是org.apache.spark.serializer.JavaSerializer。
      * 而我们要修改成org.apache.spark.serializer.KryoSerializer。
      */
    // RDD序列化到磁盘,worker与worker之间的数据传输。
    //conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //val sc = new SparkContext(conf)

    /**
      * https://www.cnblogs.com/yanshw/p/11975572.html
      * 声明：SQLContext 已在 2.x 被废除，建议使用 SparkSession
      * 1.
      * spark 有三大引擎，spark core、sparkSQL、sparkStreaming，
      * spark core 的关键抽象是 SparkContext、RDD；
      * SparkSQL 的关键抽象是 SparkSession、DataFrame；
      * sparkStreaming 的关键抽象是 StreamingContext、DStream
      * 2.
      * SparkSession 是 spark2.0 引入的概念，主要用在 sparkSQL 中，当然也可以用在其他场合，他可以代替 SparkContext；
      * SparkSession 其实是封装了 SQLContext 和 HiveContext
      * 3.
      * SparkContext 是什么?
      *   驱动程序使用SparkContext与集群进行连接和通信，它可以帮助执行Spark任务，并与资源管理器(如YARN 或Mesos)进行协调。
      *   使用SparkContext，可以访问其他上下文，比如SQLContext和HiveContext。
      *   使用SparkContext，我们可以为Spark作业设置配置参数。
      */
    /**
      * 1.SQLContect：SQLContext是通往SparkSQL的入口，sparkSQL 的应用必须创建一个 SQLContext 或者 HiveContext 的类实例
      * 2.HiveContext:HiveContext是通往hive入口。，它继承自 SQLContext，用于处理 hive 中的数据
      *     HiveContext 对 SQLContext 进行了扩展，功能要强大的多
      *   (1). 它可以执行 HiveSQL 和 SQL 查询
      *   (2). 它可以操作 hive 数据，并且可以访问 HiveUDF
      *   (3). 它不一定需要 hive，在没有 hive 环境时也可以使用 HiveContext
      * 注意，如果要处理 hive 数据，需要把 hive 的 hive-site.xml 文件放到 spark/conf 下，HiveContext 将从 hive-site.xml 中获取 hive 配置信息；
      * 如果 HiveContext 没有找到 hive-site.xml，他会在当前目录下创建 spark-warehouse 和 metastore_db 两个文件夹
      * eg：new HiveContext(sc).sql('''select * from hive1101.person limit 2''')
      *
      * 3.SparkSession:它实现了对上面二者的封装。创建SparkSession的三种方式：
      * ### method 1
      * sess = SparkSession.builder \
      * .appName("aaa") \
      * .config("spark.driver.extraClassPath", sparkClassPath) \
      * .master("local") \
      * .enableHiveSupport() \  # sparkSQL 连接 hive 时需要这句
      * .getOrCreate()      # builder 方式必须有这句
      *
      * ### method 2
      * conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 设定 appname 和 master
      * sess = SparkSession.builder.config(conf=conf).getOrCreate() # builder 方式必须有这句
      *
      * ### method 3
      * from pyspark import SparkContext, SparkConf
      * conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 设定 appname 和 master
      * sc = SparkContext(conf=conf)
      * sess = SparkSession(sc)
      */

    // 2.创建2.0版本的sparkSession
    val sparkSession: SparkSession = SparkSession.builder().appName("ProCityRpt").master("local[*]")
      .enableHiveSupport() //支持sql
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")//spark优化：spark.serializer修改磁盘序列化方式
      .getOrCreate()

    // 3.使用Sql读取 parquet文件数据
    val df: DataFrame = sparkSession.read.parquet(logInputPath)

    // 4.将DataFrame注册成一张临时表。注意：df已经不用注册表了，而是用注册临时视图啦！！。
    df.createOrReplaceTempView("log")

    // 5.按照省市进行分组聚合 => 统计分组后的各省市的日志记录条数。
    val res: DataFrame = sparkSession.sql("select Provincename, Cityname, count(*) as ct from log group by Provincename,Cityname ")

    /**
      * 使用hadoop的文件对象的方法来判断本地存储路径是否存在，如果存在则删除。
      * 注意虽然这里用的是hadoop判断目录存在的方法，但是我们输入的路径是本地目录的而不是hdfs://master:9000这种的，
      */
    // 6.判断存储路径是否存在，如果存在则删除。
    val sc: SparkContext = sparkSession.sparkContext//使用sparkSession.sparkContext获取socket。
    val hadoopConfiguration: Configuration = sc.hadoopConfiguration
    val fs: FileSystem = FileSystem.get(hadoopConfiguration)
    val resultPath = new Path(resultOutputPath)
    if (fs.exists(resultPath)){
      fs.delete(resultPath, true)
    }

    /**
      * 7.将查询结果存储成json文件。
      * res.write.json(resultOutputPath)
      * 因为上面那样输出的话文件太多了，因此可以合并分区coalesce(1)#这里的1代表合并成了一个分区。
      * .mode(SaveMode.Overwrite)以为存储方式SavaMode以为Overwrite重写的方式保存到数据库中，SaveMode.Append是追加的方式保存到数据库中
      */
    res.write.mode(SaveMode.Overwrite).json(resultOutputPath)


    // 8.关闭sparkSession
    sparkSession.stop()


    //本地测试运行：!!!!!注意提前设置好参数：输入目录和输出目录
    /**
      * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\biz2parquet
      * 输出目录: E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\procityrpt
      *
      */

  }

}
